package simpleflink;

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class RichFunctionTest {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置并行度为1，结果输出两次close，
        env.setParallelism(1);

        DataStream<String> dataStream = env.readTextFile("data/user.txt");

        DataStream<String> mapStream = dataStream.map(new MyRichMapperFunction());

        mapStream.print();

        env.execute();
    }

    //第一个参数是输入值，第二个参数是输出
    public static class MyRichMapperFunction extends RichMapFunction<String,String>{
        @Override
        public String map(String s) throws Exception {
            return "love_"+s;
        }

        @Override
        public void open(Configuration parameters) throws Exception {
            // 初始化工作，一般是定义状态，或者建立数据库连接
            System.out.println("open");
        }

        @Override
        public void close() throws Exception {
            // 一般是关闭连接和清空状态的收尾操作
            System.out.println("close---");
        }
    }
}
